package com.hou.rocketmq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.Arrays;

/**
 * DefaultMQPushConsumer:
 *   由系统控制读取操作,收到消息后自动调用传入的处理方法来处理
 */
public class DefaultMQPushConsumerTest {
    public static void main(String[] args) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("houzheng"); //group名称
        consumer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");// 设置nameServer地址，多个以;分隔
        /**
         * 1. CONSUME_FROM_LAST_OFFSET：第一次启动从队列最后位置消费，后续再启动接着上次消费的进度开始消费
           2. CONSUME_FROM_FIRST_OFFSET：第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费
           3. CONSUME_FROM_TIMESTAMP：第一次启动从指定时间点位置消费，后续再启动接着上次消费的进度开始消费
           以上所说的第一次启动是指从来没有消费过的消费者，如果该消费者消费过，那么会在broker端记录该消费者的消费位置，如果该消费者挂了再启动，那么自动从上次消费的进度开始
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); //指定从队列哪里开始消费
        /**
         * CLUSTERING：默认模式，同一个ConsumerGroup(groupName相同)每个consumer只消费所订阅消息的一部分内容，
         *                      同一个ConsumerGroup里所有的Consumer消息加起来才是订阅topic整体，从而达到负载均衡的目的
         * BROADCASTING：同一个ConsumerGroup每个consumer都消费到所订阅topic所有消息，也就是一个消费会被多次分发，被多个consumer消费。
         */
        consumer.setMessageModel(MessageModel.BROADCASTING); // 配合group一起使用
        // 订阅topic，可以利用tag对指定消息进行过滤，例如："TopicTest","tagl||tag2||tag3",表示消费topic下面带有tag等的消息,*或null表示topic所有消息
        try {
            consumer.subscribe("test-topic-01", "*"); // tag是在发送消息时设置的标签
            consumer.registerMessageListener((MessageListenerConcurrently) (mgs, consumeconcurrentlycontext) -> {
                mgs.forEach(x->{
                    final String s = new String(x.getBody());
                    System.out.println(s);
                });
                // ConsumeConcurrentlyStatus.RECONSUME_LATER boker会根据设置的messageDelayLevel发起重试，默认16次
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
